package com.test.finkcdc.util;

import com.test.finkcdc.entity.JsonRecord;
import io.debezium.data.Envelope;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Map;

public class OutputTableSubProcessFun extends ProcessFunction<JsonRecord,Object> {
    private final OutputTag<JsonRecord> create;
    private final OutputTag<JsonRecord> update;
    private final OutputTag<JsonRecord> delete;

    public OutputTableSubProcessFun(OutputTag<JsonRecord> create,OutputTag<JsonRecord> update,OutputTag<JsonRecord> delete) {
        this.create = create;
        this.update = update;
        this.delete = delete;
    }

    public void processElement(JsonRecord jsonRecord, ProcessFunction<JsonRecord, Object>.Context ctx, Collector<Object> collector) throws Exception {
        Envelope.Operation op=jsonRecord.getOp();
        if(op==Envelope.Operation.CREATE){
            ctx.output(create, jsonRecord);
        }else if(op==Envelope.Operation.UPDATE){
            ctx.output(update, jsonRecord);

        }else if(op==Envelope.Operation.UPDATE){
            ctx.output(delete, jsonRecord);
        }
    }
}